package com.example.kafka;

import com.example.enums.EventType;
import com.example.kafka.data.BaseEventData;
import com.example.kafka.handler.EventHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Component
public class EventHandlerFactory {

    private static final Logger logger = LoggerFactory.getLogger(EventHandlerFactory.class);
    private final Map<Integer, EventHandler<? extends BaseEventData>> handlerMap = new ConcurrentHashMap<>();

    public EventHandlerFactory(List<EventHandler<? extends BaseEventData>> handlers) {
        for (EventHandler<? extends BaseEventData> handler : handlers) {
            handlerMap.put(handler.getEventType().getEvenType(), handler);
        }
    }

    public EventHandler<? extends BaseEventData> getHandlerForEventType(Integer eventType) {
        return handlerMap.get(eventType);
    }

    public Set<EventType> getAllEventTypes() {
        return handlerMap.keySet().stream()
                .map(EventType::of)
                .collect(Collectors.toSet());
    }

    public <T extends BaseEventData> void processEvent(KafkaEvent event) {
        @SuppressWarnings("unchecked")
        EventHandler<T> handler = (EventHandler<T>) getHandlerForEventType(event.getEventType());
        if (handler != null) {
            if (handler.validateEvent(event)) {
                handler.handleEvent(event);
            } else {
                logger.warn("Event validation failed for event type: {}", event.getEventType());
            }
        } else {
            logger.warn("No handler found for event type: {}", event.getEventType());
        }
    }

}
